package com.shujia.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object WordCount {
  def main(args: Array[String]): Unit = {
    //构建Spark sql的执行环境
    val ss: SparkSession = SparkSession.builder()
      .master("local")
      .appName("spark sql单词统计")
      .getOrCreate()

    //spark sql是spark core，对Spark core一些功能做了封装，直接使用SparkSession获取Spark core的运行环境
    val sc: SparkContext = ss.sparkContext

    //-----------------------------------------------------

    //使用SparkSession读取本地文件，封装成DataFrame类型的数据
    val df1: DataFrame = ss.read
      .format("csv") // 我们这里传入的是csv,但是实际读取文本数据的列分隔符不一定非要是英文逗号,与option方法结合使用，读取实际的分割数据
      .option("sep", "\n") // 指定列与列之间的分隔符
      .schema("line STRING") // 构建DataFrame的字段结构，包含字段名和数据类型。兼容hive数据类型
      .load("spark/data/words.txt")

//    df1.show() //查看DataFrame内容

    /**
     *  无法直接在DataFrame的数据结构上编写sql语句
     *  需要将一个DataFrame对象映射成一张虚拟的表，然后才可以编写sql语句进行数据分析
     */
    // 老版本的做法
//    df1.registerTempTable("wc")
    // 新版本的做法
    df1.createOrReplaceTempView("wc")

    /**
     *  使用SparkSession对象执行sql语句
     *  sql语句中的函数，完全支持hive所有的内置函数
     */
//    val df2: DataFrame = ss.sql("select line,'数加' as tmp_col from wc")
//    df2.show()

    val resDF: DataFrame = ss.sql(
      """
        |select
        |t1.word as word,
        |count(1) as `counts`,
        |'方直' as tmp_col
        |from
        |(
        |select
        |explode(split(line,' ')) as word
        |from
        |wc) t1
        |group by t1.word
        |""".stripMargin)


    /**
     *  将结果保存在目录里
     */
    resDF.repartition(1) // 重分区，将分区设置为1个
      .write
      .format("csv")
      .option("sep","\\|")
      .mode(SaveMode.Overwrite) // 覆盖写入结果文件中
      .save("spark/output/out4")


  }
}
